package orderdata;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Date;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

public class OrderDataSource  implements SourceFunction<OrderData> {
    // 声明标识位，用于控制数据源的运行状态
    private Boolean running = true;

    @Override
    public void run(SourceContext<OrderData> sourceContext) throws Exception {
        // 初始化订单 ID
        long orderId = 0;
        // 随机数生成器
        Random random = new Random();
        // 商品品类列表
        String[] categories = {"Electronics", "Clothing", "Books", "Furniture", "Food"};
        // 销售地区列表
        String[] regions = {"North America", "Europe", "Asia", "Africa", "Australia"};
        // 循环生成数据，直到 cancel() 被调用
        while (running) {
            // 生成随机时间戳
            long randomMillis = ThreadLocalRandom.current().nextLong(
                    new Date(2000 - 1900, 0, 1).getTime(),
                    new Date(2025 - 1900, 11, 31).getTime()
            );

            // 创建一个订单对象并发送到下游
            sourceContext.collect(new OrderData(
                    randomMillis, // 订单日期
                    categories[random.nextInt(categories.length)], // 商品品类
                    regions[random.nextInt(regions.length)], // 销售地区
                    100 + random.nextInt(1000), // 用户 ID
                    1 + random.nextInt(5), // 订单数
                    50 + random.nextDouble() * 450 // 订单金额（50 到 500 之间）
            ));

            // 控制数据生成速率，避免过快生成
            Thread.sleep(1000); // 每秒生成一条数据
        }
    }

    @Override
    public void cancel() {
        // 设置标识位为 false，终止数据生成
        running = false;
    }
}
